package com.globalwave.datasync.engine;

import java.sql.BatchUpdateException;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Set;

import com.globalwave.common.db.ConnectionProvider;

public class MoveSync extends BaseBatchSync {


	
	public MoveSync(String sourceDbCode, String syncConfigCode,
			ConnectionProvider connectionProvider) {
		super(sourceDbCode, syncConfigCode, connectionProvider);
	}

    /**
     * 数据同步处理，把数据集中的内容，同步到目标数据库
     * 
     * @param executeSQL 将要执行的SQL
     * @param srcResultSet 原数据集，只有一个数据记录，并当前指向该记录
     * @param dbKeyValues 数据库key值
     * 
     */
    protected void executeBatchInsert(ResultSet srcResultSet, Set<Long> existIds, Long start, Long end, int size) throws Exception {
        String sql = null;
        Object[] ids = null ;// 用于出错时查找出错的ID
        try {			
        	final boolean isExistEmpty = (existIds.size() == 0) ;
        	final String fieldName = config.getSource().getKeys()[0].getSyncColumnsCode() ;
        	
        	sql = config.getDestinationInsertSQL() ;
        	
        	PreparedStatement stmt = getPrepareStatement(destConnection, sql);
                   
            //ResultSetMetaData meta = srcResultSet.getMetaData();
            //int relationLen = meta.getColumnCount() ;

            ids = new Object[size] ;
            destConnection.setAutoCommit(false) ;
            
            while (srcResultSet.next()) {
            	Object id = srcResultSet.getLong(fieldName) ;
            	ids[srcResultSet.getRow() - 1] = id ;
            	if (isExistEmpty || !existIds.contains(id)) {
	                copyOneRowData(srcResultSet, stmt);
	                stmt.addBatch() ;
            	}
            }
	        
            int[] result = stmt.executeBatch();

        	if (log.isDebugEnabled()) {
                log.debug(" execute result : " + result);
        	}
        	
            srcConnection.setAutoCommit(false) ;
            ids = new Object[existIds.size() + 1] ;
            
            sql = config.getSourceDeleteSQL() ;
            int i = 0 ; 
        	ids[i++] = start + "~" + end ;
        	
        	PreparedStatement updateFlagStm = getPrepareStatement(srcConnection, sql);
        	updateFlagStm.setLong(1, start) ;
        	updateFlagStm.setLong(2, end) ;
        	updateFlagStm.execute() ;

        	if (log.isDebugEnabled()) {
                 log.debug(" execute result : " + result);
        	}
        	
            srcConnection.commit() ;
            destConnection.commit() ;

        } catch(BatchUpdateException be) {
        	log.error("current execute sql:" + sql) ;
        	destConnection.rollback() ;
        	srcConnection.rollback() ;
        	processUpdateCounts(be.getUpdateCounts(), ids, "id is[%s]") ;
            throw be;
        } catch(SQLException ex) {
        	log.error("current execute sql:" + sql) ;
        	destConnection.rollback() ;
        	srcConnection.rollback() ;
            throw ex;
        } catch(Exception e) {
        	destConnection.rollback() ;
        	srcConnection.rollback() ;
            throw e;
        } 

    }
    
    
	  
}
